package cn.itcast.flink.sql;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.lit;

/**
 * Author itcast
 * Date 2022/1/18 17:24
 * Desc 随机每秒钟生成用户订单信息并求出订单金额最大、最小值和订单总笔数
 */
public class CustomOrderCountDemo {
    public static void main(String[] args) throws Exception {
        //1.准备环境 创建流执行环境和流表环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //开启checkpoint及其参数
        env.enableCheckpointing(1000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/flink-chks");
        //开启重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 3000));
        //2.Source 自定义Order 每一秒中睡眠一次
        DataStreamSource<Order> source = env.addSource(new CustomOrder());
        //3.Transformation 分配时间戳和水印2秒
        SingleOutputStreamOperator<Order> watermarkStream = source.assignTimestampsAndWatermarks(
                WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((element, timestamp) -> element.createTime)
        );

        //创建环境配置和表环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
               .inBatchMode()
                .build();
        //获取流表对象
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);
        //4.注册表 创建临时视图并分配 rowtime
        // 列模式  Alt + 鼠标左键  ，全选 ctrl + shift + -> ，选中 ctrl + ->
        tEnv.createTemporaryView(
                "t_order",
                watermarkStream,
                $("oid"),
                $("uid"),
                $("money"),
                $("createTime").rowtime()  //当前字段当成 event time
        );

        //5.编写FlinkSQL，根据 userId 和 createTime 滚动分组统计 userId、订单总笔数、最大、最小金额
        /*Table result = tEnv.sqlQuery(
                "select uid,count(oid) as cnt,max(money) as maxMoney,min(money) as minMoney " +
                        "from t_order " +
                        "group by uid, tumble(createTime, interval '5' second)"
        );*/
        //编写 FlinkTable from -> table
        Table order = tEnv.from("t_order");
        // window 开窗 -> Tumble -> lit 推断出来类型
        Table resultTable = order.window(Tumble.over(lit(5).second()).on($("createTime")).as("tumbleWindow"))
                .groupBy($("uid"), $("tumbleWindow"))
                .select($("uid"),
                        $("oid").count().as("cnt"),
                        $("money").max().as("maxMoney"),
                        $("money").min().as("minMoney")
                );


        //6.执行查询语句返回结果
        //7.Sink toRetractStream  → 将计算后的新的数据在DataStream原数据的基础上更新true或是删除false
        DataStream<Tuple2<Boolean, OrderInfo>> orderResult = tEnv.toRetractStream(resultTable, OrderInfo.class);
        //8.打印输出
        orderResult.print();
        //9.执行流环境
        env.execute();
    }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class OrderInfo {
        private String uid;
        private Long cnt;
        private Double maxMoney;
        private Double minMoney;
    }

    //自定义实现 Order ，乱序生成订单数据
    public static class CustomOrder implements SourceFunction<Order> {

        boolean isRunning = true;
        //随机数
        Random rm = new Random();

        @Override
        public void run(SourceContext<Order> ctx) throws Exception {
            while (isRunning) {
                Order order = new Order(
                        UUID.randomUUID().toString(),
                        rm.nextInt(5) + "",
                        rm.nextDouble() * 100,
                        System.currentTimeMillis() - rm.nextInt(5) * 1000
                );
                //将订单进行生成
                ctx.collect(order);
                //一秒生成
                TimeUnit.SECONDS.sleep(1);
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }

    //lombok 实现 getter setter constructor
    @AllArgsConstructor
    @NoArgsConstructor
    @Data
    public static class Order {
        private String oid;
        //用户id，是 5 编号的人
        private String uid;
        //0~100的值
        private Double money;
        //当前时间 - 随机数 5
        private Long createTime;
    }
}
